package itemCF.step1;


import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

/**
 * @author legolas
 * @date 2020/3/13 下午7:06
 */

public class Reducer1 extends Reducer<Text, Text, Text, Text> {

    /**
     * 需要实现的reduce函数
     * 将map分组后的k2，v2s进行聚合处理
     *
     * @param k2
     * @param v2s
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */

    @Override
    protected void reduce(Text k2, Iterable<Text> v2s, Context context) throws IOException, InterruptedException {
        Map<String, Integer> map = new HashMap<>();
        for (Text v2 : v2s) {
            String userID = v2.toString().split("_")[0];
            String score = v2.toString().split("_")[1];
            if (map.get(userID) == null) {
                map.put(userID, Integer.parseInt(score));
            } else {
                Integer preScore = map.get(userID);
                map.put(userID, preScore + Integer.parseInt(score));
            }
        }
        StringBuilder sb = new StringBuilder();
        for (Map.Entry<String, Integer> entry : map.entrySet()) {
            String userID = entry.getKey();
            String score = String.valueOf(entry.getValue());
            sb.append(userID).append("_").append(score).append(",");
        }
        String line = null;
        if (sb.toString().endsWith(",")) {
            line = sb.substring(0, sb.length() - 1);
        }
        Text k3 = new Text();
        k3.set(k2);
        Text v3 = new Text();
        assert line != null;//等同与if(line!=null)
        v3.set(line);
        context.write(k3, v3);

    }
}
